package com.fbs.job.admin.core.thread;

import com.fbs.job.admin.core.complete.FbsJobCompleter;
import com.fbs.job.admin.core.config.FbsJobAdminConfig;
import com.fbs.job.admin.core.entity.FbsJobLog;
import com.fbs.job.core.biz.model.HandleCallbackParam;
import com.fbs.job.core.model.ResultT;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

public class JobCompleteHelper {

    private static Logger logger = LoggerFactory.getLogger(JobCompleteHelper.class);
    private static JobCompleteHelper instance = new JobCompleteHelper();
    public static JobCompleteHelper getInstance(){
        return instance;
    }


    private ThreadPoolExecutor callbackThreadPool = null;
    private Thread monitorThread;
    private volatile boolean toStop = false;
    public void start(){
        callbackThreadPool=new ThreadPoolExecutor(2,
                20,
                30L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(3000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "fbs-job, admin JobLosedMonitorHelper-callbackThreadPool-" + r.hashCode());
                    }
                },
                new RejectedExecutionHandler() {
                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        r.run();
                        logger.warn(">>>>>>>>>>> fbs-job, callback too fast, match threadpool rejected handler(run now).");

                    }
                }
        );



        monitorThread = new Thread(new Runnable() {
            @Override
            public void run() {
                // wait for JobTriggerPoolHelper-init
                try {
                    TimeUnit.MILLISECONDS.sleep(50);
                } catch (InterruptedException e) {
                    if (!toStop) {
                        logger.error(e.getMessage(), e);
                    }
                }

                while (!toStop){
                   try {
                       // 任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败；
                       Date losedTime = DateUtils.addMinutes(new Date(), -10);
                       List<Long> losedJobIds  = FbsJobAdminConfig.getAdminConfig().getFbsJobLogMapper().findLostJobIds(losedTime);
                       if (CollectionUtils.isNotEmpty(losedJobIds)){
                           for (Long logId: losedJobIds) {

                               FbsJobLog jobLog = new FbsJobLog();
                               jobLog.setId(logId);

                               jobLog.setHandleTime(new Date());
                               jobLog.setHandleCode(ResultT.FAIL_CODE);
                               jobLog.setHandleMsg("任务结果丢失，标记失败");

                               FbsJobCompleter .updateHandleInfoAndFinish(jobLog);
                           }
                       }
                   }catch (Exception e){
                       if (!toStop) {
                           logger.error(">>>>>>>>>>> fbs-job, job fail monitor thread error:{}", e);
                       }
                   }
                    try {
                        TimeUnit.SECONDS.sleep(60);
                    } catch (Exception e) {
                        if (!toStop) {
                            logger.error(e.getMessage(), e);
                        }
                    }

                 }
                //
                logger.info(">>>>>>>>>>> fbs-job, JobLosedMonitorHelper stop");
            }
        });
        monitorThread.setDaemon(true);
        monitorThread.setName("fbs-job, admin JobLosedMonitorHelper");
        monitorThread.start();
    }

    public void toStop(){
        toStop = true;

        // stop registryOrRemoveThreadPool
        callbackThreadPool.shutdownNow();

        // stop monitorThread (interrupt and wait)
        monitorThread.interrupt();
        try {
            monitorThread.join();
        } catch (InterruptedException e) {
            logger.error(e.getMessage(), e);
        }
    }


    // ---------------------- helper ----------------------

    public ResultT<String> callback(List<HandleCallbackParam> callbackParamList) {

        callbackThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (HandleCallbackParam handleCallbackParam: callbackParamList) {
                    ResultT<String> callbackResult = callback(handleCallbackParam);
                    logger.debug(">>>>>>>>> JobApiController.callback {}, handleCallbackParam={}, callbackResult={}",
                            (callbackResult.getCode()== ResultT.SUCCESS_CODE?"success":"fail"), handleCallbackParam, callbackResult);
                }
            }
        });

        return ResultT.success();
    }

    private ResultT<String> callback(HandleCallbackParam handleCallbackParam) {
        // valid log item
        FbsJobLog log = FbsJobAdminConfig.getAdminConfig().getFbsJobLogMapper().getById(handleCallbackParam.getLogId());
        if (log == null) {
            return new ResultT<String>(ResultT.FAIL_CODE, "log item not found.");
        }
        if (log.getHandleCode() > 0) {
            return new ResultT<String>(ResultT.FAIL_CODE, "log repeate callback.");     // avoid repeat callback, trigger child job etc
        }

        // handle msg
        StringBuffer handleMsg = new StringBuffer();
        if (log.getHandleMsg()!=null) {
            handleMsg.append(log.getHandleMsg()).append("<br>");
        }
        if (handleCallbackParam.getHandleMsg() != null) {
            handleMsg.append(handleCallbackParam.getHandleMsg());
        }

        // success, save log
        log.setHandleTime(new Date());
        log.setHandleCode(handleCallbackParam.getHandleCode());
        log.setHandleMsg(handleMsg.toString());
        FbsJobCompleter.updateHandleInfoAndFinish(log);

        return ResultT.success();
    }
}
